package t20240516_lineage;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.*;


public class Test3 {

    public static void main(String[] args) {


        Configuration configuration = new Configuration();
        configuration.setBoolean("table.exec.disable-operator-chaining", true);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(configuration);
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);

        String sql ="create table o_scdn_vod_xserver_monitor_cycle (\n" +
                "    `stime`\tstring,\n" +
                "    `host`\tstring,\n" +
                "    `node_level` int,\n" +
                "    `domain`\tstring,\n" +
                "    `req_http_cnt`\tbigint,\n" +
                "    `req_https_cnt`  bigint,\n" +
                "    `req_v4_cnt`  bigint,\n" +
                "    `req_v6_cnt`  bigint,\n" +
                "    `source_cnt`  bigint,\n" +
                "    `http_flow`  bigint,\n" +
                "    `https_flow`  bigint,\n" +
                "    `v4_flow`  bigint,\n" +
                "    `v6_flow`  bigint,\n" +
                "    `source_http_flow`  bigint,\n" +
                "    `source_https_flow`  bigint,\n" +
                "    `http_hit_cnt`  bigint,\n" +
                "    `https_hit_cnt`  bigint,\n" +
                "    `http_hit_flow`  bigint,\n" +
                "    `https_hit_flow`  bigint,\n" +
                "    `code_detail`  string,\n" +
                "    `source_code_detail`  string,\n" +
                "    rowtime as to_timestamp(`stime`),\n" +
                "    proctime as proctime(),\n" +
                "    watermark for rowtime as rowtime - interval '30' second\n" +
                ") with (\n" +
                "    'properties.bootstrap.servers' = 'bdyun-suzhou-tw04a0264.su.baidu.internal:6667,bdyun-suzhou-tw04a0265.su.baidu.internal:6667,bdyun-suzhou-tw04d0849.su.baidu.internal:6667',\n" +
                "    'connector' = 'kafka',\n" +
                "    'format' = 'avro-event',\n" +
                "    'properties.group.id' = 'flink-swallow-o_scdn_vod_xserver_monitor_cycle_statv3-swallow',\n" +
                "    'topic' = 'o_scdn_vod_xserver_monitor_cycle',\n" +
                "    'properties.security.protocol' = 'SASL_PLAINTEXT',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'properties.sasl.kerberos.service.name' = 'kafka'\n" +
                ");\n" +
                "\n" +
                "-- create table o_scdn_vod_ali_cdn_realtime_log (\n" +
                "--     `stime` string,\n" +
                "--     `domain`  string,\n" +
                "--     `method`    string,\n" +
                "--     `scheme`  string,\n" +
                "--      `uri`              string,  \n" +
                "--      `uri_param`        string,  \n" +
                "--      `client_ip`        string,  \n" +
                "--      `remote_ip`        string,  \n" +
                "--      `remote_port`      int,   \n" +
                "--      `request_size`     bigint,   \n" +
                "--      `request_time`     double, \n" +
                "--      `response_size`    bigint,   \n" +
                "--      `return_code`      int,   \n" +
                "--     rowtime as to_timestamp(`stime`),\n" +
                "--     proctime as proctime(),\n" +
                "--     watermark for rowtime as rowtime - interval '60' second\n" +
                "-- ) with (\n" +
                "--     'properties.bootstrap.servers' = 'bdyun-suzhou-tw04a0264.su.baidu.internal:6667,bdyun-suzhou-tw04a0265.su.baidu.internal:6667,bdyun-suzhou-tw04d0849.su.baidu.internal:6667',\n" +
                "--     'connector' = 'kafka',\n" +
                "--     'format' = 'avro-event',\n" +
                "--     'properties.group.id' = 'flink-swallow-o_scdn_vod_ali_cdn_realtime_log_statv3-swallow',\n" +
                "--     'topic' = 'o_scdn_vod_ali_cdn_realtime_log',\n" +
                "--     'properties.security.protocol' = 'SASL_PLAINTEXT',\n" +
                "--     'scan.startup.mode' = 'latest-offset',\n" +
                "--     'properties.sasl.kerberos.service.name' = 'kafka'\n" +
                "-- );\n" +
                "\n" +
                "-- create table o_scdn_vod_bd_cdn_realtime_log (\n" +
                "--    `stime` string,\n" +
                "--     `domain`  string,\n" +
                "--     `req_method`    string,\n" +
                "--      `uri`              string,  \n" +
                "--      `parameters`        string,  \n" +
                "--      `remote_ip`        string,  \n" +
                "--      `request_length`   bigint,    \n" +
                "--      `request_time`     double, \n" +
                "--      `bytes_sent`    bigint,   \n" +
                "--      `status`      int,\n" +
                "--     rowtime as to_timestamp(`stime`),\n" +
                "--     proctime as proctime(),\n" +
                "--     watermark for rowtime as rowtime - interval '60' second\n" +
                "-- ) with (\n" +
                "--     'properties.bootstrap.servers' = 'bdyun-suzhou-tw04a0264.su.baidu.internal:6667,bdyun-suzhou-tw04a0265.su.baidu.internal:6667,bdyun-suzhou-tw04d0849.su.baidu.internal:6667',\n" +
                "--     'connector' = 'kafka',\n" +
                "--     'format' = 'avro-event',\n" +
                "--     'properties.group.id' = 'flink-swallow-o_scdn_vod_bd_cdn_realtime_log_statv3-swallow',\n" +
                "--     'topic' = 'o_scdn_vod_bd_cdn_realtime_log',\n" +
                "--     'properties.security.protocol' = 'SASL_PLAINTEXT',\n" +
                "--     'scan.startup.mode' = 'latest-offset',\n" +
                "--     'properties.sasl.kerberos.service.name' = 'kafka'\n" +
                "-- );\n" +
                "\n" +
                "create table t_domain_of_data_platform (\n" +
                "\t`userid`\tbigint,\n" +
                "\t`domain_name`\tstring\n" +
                ") with (\n" +
                "\t'connector' = 'tidb',\n" +
                "\t'driver' = 'com.mysql.jdbc.Driver',\n" +
                "\t'lookup.cache.ttl' = '30min',\n" +
                "\t'table-name' = 't_domain_of_data_platform',\n" +
                "\t'url' = 'jdbc:mysql://tidb-lb.data.onething.internal:4000/dim?useUnicode=true&useSSL=false&characterEncoding=UTF-8&allowMultiQueries=true&autoReconnect=true',\n" +
                "\t'lookup.max-retries' = '3',\n" +
                "\t'username' = 'beta',\n" +
                "    'password' = 'beta@9898w',\n" +
                "\t'query-sql' = 'select account_id as userid, domain_name from t_domain_of_data_platform'\n" +
                ");\n" +
                "\n" +
                "-- create table o_scdn_vod_ty_cdn_realtime_log (\n" +
                "-- \t`stime`\t\t\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`body_bytes_sent`\t\t\t\t\t\tbigint\t,\n" +
                "-- \t`bytes_sent`\t\t\t\t\t\t\tbigint\t,\n" +
                "-- \t`cache_status`\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`domain`\t\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`filesize`\t\t\t\t\t\t\t\tbigint\t,\n" +
                "-- \t`http_user_agent`\t\t\t\t\t\tstring\t,\n" +
                "-- \t`http_referer`\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`http_range`\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`parameters`\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`request_time`\t\t\t\t\t\t\tdouble\t,\n" +
                "-- \t`req_method`\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`request_length`\t\t\t\t\t\tbigint\t,\n" +
                "-- \t`remote_ip`\t\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`response_body_first_byte_time`\t\t\tdouble\t,\n" +
                "-- \t`request_id`\t\t\t\t\t\t\tstring\t,\n" +
                "-- \t`request_header_time`\t\t\t\t\tstring\t,\n" +
                "-- \t`status`\t\t\t\t\t\t\t\tint\t,\n" +
                "--     rowtime as to_timestamp(`stime`),\n" +
                "--     proctime as proctime(),\n" +
                "--     watermark for rowtime as rowtime - interval '60' second\n" +
                "-- ) with (\n" +
                "--     'properties.bootstrap.servers' = 'bdyun-suzhou-tw04a0264.su.baidu.internal:6667,bdyun-suzhou-tw04a0265.su.baidu.internal:6667,bdyun-suzhou-tw04d0849.su.baidu.internal:6667',\n" +
                "--     'connector' = 'kafka',\n" +
                "--     'format' = 'avro-event',\n" +
                "--     'properties.group.id' = 'flink-swallow-o_scdn_vod_ty_cdn_realtime_log-swallow',\n" +
                "--     'topic' = 'o_scdn_vod_ty_cdn_realtime_log',\n" +
                "--     'properties.security.protocol' = 'SASL_PLAINTEXT',\n" +
                "--     'scan.startup.mode' = 'latest-offset',\n" +
                "--     'properties.sasl.kerberos.service.name' = 'kafka'\n" +
                "-- );\n" +
                "create table d_scdn_mix_cdn_domain (\n" +
                "\t  `domain`    string,\n" +
                "      `config_domain` string\n" +
                ") with (\n" +
                "\t'connector' = 'tidb',\n" +
                "\t'driver' = 'com.mysql.jdbc.Driver',\n" +
                "    'table-name' = 'dim_scdn_mix_cdn_domain',\n" +
                "\t'lookup.cache.ttl' = '30min',\n" +
                "\t'url' = 'jdbc:mysql://tidb-lb.data.onething.internal:4000/dim?useSSL=false&useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&autoReconnect=true&failOverReadOnly=false&maxReconnects=3&queryTimeout=2400',\n" +
                "\t'lookup.max-retries' = '3',\n" +
                "    'username' = 'beta',\n" +
                "    'password' = 'beta@9898w',\n" +
                "    'query-sql' = 'select domain, config_domain from dim_scdn_mix_cdn_domain where domain <> config_domain group by domain,config_domain'\n" +
                ");\n" +
                "create table app_i1_scdn_vod_xserver_httpcode_stat (\n" +
                "\t`stime`\ttimestamp,\n" +
                "\t`grpid`\tint,\n" +
                "\t`account_id`\tstring,\n" +
                "\t`domain`\tstring,\n" +
                "\t`hostname`\tstring,\n" +
                "\t`level`\tstring,\n" +
                "\t`status`\tint,\n" +
                "\t`status_cnt_total`\tbigint\n" +
                ") with (\n" +
                "\t'database' = 'scdn_app_lo',\n" +
                "\t'password' = 'rNXB>=dQCe3Ri-CGg*B?',\n" +
                "\t'table-mode' = 'single',\n" +
                "\t'sink.write-mode' = 'local',\n" +
                "\t'connector' = 'clickhouse',\n" +
                "\t'sink.flush-interval' = '10s',\n" +
                "\t'sink.batch-size' = '10000',\n" +
                "\t'url' = 'clickhouse://ck-lb.data.onething.internal:8123',\n" +
                "\t'table' = 'app_i1_scdn_vod_xserver_httpcode_stat',\n" +
                "\t'username' = 'spark'\n" +
                ");\n" +
                "\n" +
                "\n" +
                "create view o_scdn_vod_xserver_monitor_cycle_filter as\n" +
                "select\n" +
                "    domain,\n" +
                "    code_detail,\n" +
                "    exploded_col,\n" +
                "    CARDINALITY(string_split(code_detail, ',')) as `code_num`,\n" +
                "    rowtime\n" +
                "    \n" +
                "from o_scdn_vod_xserver_monitor_cycle,\n" +
                "     LATERAL TABLE(split(code_detail, ',')) AS T(exploded_col)\n" +
                "where code_detail like '%:%'\n" +
                "                    AND node_level = 3\n" +
                "                    and code_detail not like '%:,%'\n" +
                "            and code_detail not like '%,:%'\n" +
                "            and left(code_detail, 1) not in ( ':',',')\n" +
                "            and right(code_detail,1)  not in (':',',') ;\n" +
                "\n" +
                "create view o_scdn_vod_xserver_monitor_cycle_filter2 as\n" +
                "select domain,\n" +
                "    code_detail,\n" +
                "    exploded_col,\n" +
                "    `code_num`,\n" +
                "    rowtime\n" +
                "     from o_scdn_vod_xserver_monitor_cycle_filter\n" +
                "where cast(split_index(exploded_col,':',0) as int) <1000;\n" +
                "\n" +
                "create view o_scdn_vod_xserver_monitor_cycle_filter_group as \n" +
                "select \n" +
                "     window_start,\n" +
                "    window_end,\n" +
                "    map_join_with_wildcard(`domain`) as field,\n" +
                "    `domain`,\n" +
                "    cast(split_index(exploded_col,':',0) as int) as `httpcode`,\n" +
                "    sum(cast(split_index(exploded_col,':',1) as bigint)) as `status_cnt_total`,\n" +
                "     proctime() as proctime\n" +
                "from table(tumble(table o_scdn_vod_xserver_monitor_cycle_filter2, descriptor(`rowtime`), interval '60' second ))\n" +
                "group by\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    `domain`,\n" +
                "    split_index(exploded_col,':',0),\n" +
                "    map_join_with_wildcard(`domain`) \n" +
                "    ;\n" +
                "\n" +
                "-- create view o_scdn_vod_ali_cdn_realtime_log_filter as\n" +
                "-- select\n" +
                "--     domain,\n" +
                "--     return_code as `httpcode`,\n" +
                "--     proctime,\n" +
                "--     rowtime\n" +
                "-- from o_scdn_vod_ali_cdn_realtime_log\n" +
                "-- where (uri_param not LIKE '%xy_push=1%' or uri_param is null) \n" +
                "-- union all\n" +
                "-- select\n" +
                "--     domain,\n" +
                "--     status as `httpcode`,\n" +
                "--     proctime,\n" +
                "--     rowtime\n" +
                "-- from o_scdn_vod_bd_cdn_realtime_log\n" +
                "-- where (parameters not LIKE '%xy_push=1%' or parameters is null) \n" +
                "-- union all\n" +
                "-- select\n" +
                "--     domain,\n" +
                "--     status as `httpcode`,\n" +
                "--     proctime,\n" +
                "--     rowtime\n" +
                "-- from o_scdn_vod_ty_cdn_realtime_log\n" +
                "-- where (parameters not LIKE '%xy_push=1%' or parameters is null) ;\n" +
                "\n" +
                "-- create view o_scdn_vod_ali_cdn_realtime_log_filter_join as\n" +
                "-- select \n" +
                "--     case when t2.domain is null then t1.domain else t2.domain end as domain,\n" +
                "--     httpcode,\n" +
                "--     rowtime,\n" +
                "--     proctime\n" +
                "-- from o_scdn_vod_ali_cdn_realtime_log_filter t1\n" +
                "-- left join d_scdn_mix_cdn_domain for system_time as of t1.proctime t2 on t1.domain = t2.config_domain;\n" +
                "\n" +
                "-- create view o_scdn_vod_ali_cdn_realtime_log_filter_group as \n" +
                "-- select \n" +
                "--     window_start,\n" +
                "--     window_end,\n" +
                "--     domain,\n" +
                "--     httpcode,\n" +
                "--     count(1) as `status_cnt_total`,\n" +
                "--     proctime\n" +
                "-- from table(tumble(table o_scdn_vod_ali_cdn_realtime_log_filter_join, descriptor(`rowtime`), interval '60' second ))\n" +
                "-- group by\n" +
                "--     window_start,\n" +
                "--     window_end,\n" +
                "--     domain,\n" +
                "--     httpcode,\n" +
                "--     proctime\n" +
                "-- union all\n" +
                "-- select \n" +
                "--      window_start,\n" +
                "--     window_end,\n" +
                "--     `domain`,\n" +
                "--      `httpcode`,\n" +
                "--     status_cnt_total,\n" +
                "--     proctime   \n" +
                "--   from   o_scdn_vod_xserver_monitor_cycle_filter_group\n" +
                "--     ;\n" +
                "\n" +
                "insert into app_i1_scdn_vod_xserver_httpcode_stat \n" +
                "select\n" +
                "    window_start as stime,\n" +
                "    1 as grpid,\n" +
                "    ifnull(split_index(t1.`field`, '|', 1), cast(userid as string)) `account_id`,\n" +
                "    `domain`,\n" +
                "    '-9999' hostname,\n" +
                "    'edge' level,\n" +
                "    httpcode as  status,\n" +
                "    status_cnt_total\n" +
                "from o_scdn_vod_xserver_monitor_cycle_filter_group t1\n" +
                "left join t_domain_of_data_platform for system_time as of t1.proctime t2 on split_index(t1.`field`, '|', 0) = t2.domain_name;";



        String[] statements = getStatements(sql);

        LineageContext lineageContext = new LineageContext(tbEnv);
        Tuple2<List<LineageRel>, LineageResult> tuple2 = lineageContext.analyzeLineage(statements);
        List<LineageRel> lineageRels = tuple2.f0;

        getTestJson(lineageRels);
        printLineage(lineageRels);

        LineageResult f1 = tuple2.f1;

        JSONObject o = (JSONObject)JSONObject.toJSON(f1);
        System.out.println();
        System.out.println(o.toJSONString());


    }

    private static void printLineage(List<LineageRel> lineageRels) {
        System.out.println("=======================================================================");
        for (LineageRel lineageRel : lineageRels) {
            String log = String.format("%s.%s -> %s.%s", lineageRel.getSourceTable(), lineageRel.getSourceColumn()
                    , lineageRel.getTargetTable(), lineageRel.getTargetColumn());
            System.out.println(log);
        }
        System.out.println("=======================================================================");

    }


    private static void getTestJson(List<LineageRel> lineageRels) {


        Map<String, Set<String>> nodes = new HashMap<>();


        // edges
        JSONArray array = new JSONArray();
        for (LineageRel rel : lineageRels) {
            JSONObject from = new JSONObject();
            from.put("name", rel.getSourceTable());
            from.put("field", rel.getSourceColumn());
            from.put("transform", rel.getTransform());
            System.out.println("transform:" + rel.getTransform());

            JSONObject to = new JSONObject();
            to.put("name", rel.getTargetTable());
            to.put("field", rel.getTargetColumn());


            addNode(nodes, rel.getSourceTable(), rel.getSourceColumn());
            addNode(nodes, rel.getTargetTable(), rel.getTargetColumn());


            JSONObject object = new JSONObject();

            object.put("from", from);
            object.put("to", to);

            array.add(object);
        }

        // nodes

        JSONArray nodeArrays = new JSONArray();
        for (Map.Entry<String, Set<String>> stringSetEntry : nodes.entrySet()) {
            String tableName = stringSetEntry.getKey();
            JSONObject node = new JSONObject();
            node.put("name", tableName);

            JSONArray fileds = new JSONArray();
            for (String s : stringSetEntry.getValue()) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("name", s);
                fileds.add(jsonObject);
            }
            node.put("fields", fileds);

            nodeArrays.add(node);

        }


        JSONObject result = new JSONObject();
        result.put("edges", array);
        result.put("nodes", nodeArrays);

        System.out.println(result.toJSONString());
    }


    private static void addNode(Map<String, Set<String>> nodes,String tableName, String field) {
        Set<String> sourceFields = nodes.get(tableName);
        if(sourceFields != null) {
            sourceFields.add(field);
        } else {
            Set<String> list = new HashSet<>();
            list.add(field);
            nodes.put(tableName, list);
        }
    }
    public static String[] getStatements(String sql) {


        final String localSqlSeparator = ";\\s*(?:\\n|--.*)";
        String[] splits = sql.replace("\r\n", "\n")
                .split(localSqlSeparator);
        String lastStatement = splits[splits.length - 1].trim();
        if (lastStatement.endsWith(";")) {
            splits[splits.length - 1] = lastStatement.substring(0, lastStatement.length() - 1);
        }

        return splits;
    }
}
